// This file is part of OpenTSDB.
// Copyright (C) 2017 TeleDB Development Group.
//
// This program is free software: you can redistribute it and/or modify it
// under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 2.1 of the License, or (at your
// option) any later version.  This program is distributed in the hope that it
// will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty
// of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU Lesser
// General Public License for more details.  You should have received a copy
// of the GNU Lesser General Public License along with this program.  If not,
// see <http://www.gnu.org/licenses/>.
package net.opentsdb.tools.http;


import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import com.google.common.cache.RemovalListener;
import com.google.common.cache.RemovalNotification;
import com.google.common.collect.ImmutableList;
import net.opentsdb.bean.MetaBean;
import net.opentsdb.utils.JSON;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * 整合一秒内所有数据发送
 *
 */
public class CompactStream {

    private List<Map<String,MetaBean>> list = ImmutableList.of(new ConcurrentHashMap<>(),new ConcurrentHashMap<>());
    private Map<String,MetaBean> currentMap;

    public CompactStream(String url){
        currentMap = list.get(0);

        ScheduledExecutorService executor = Executors.newScheduledThreadPool(1);
        executor.scheduleWithFixedDelay(
                () -> {
                    for(int i=0; i<2; i++){
                        Map<String,MetaBean> map = list.get(i);
                        if(!map.isEmpty()){
                            currentMap = list.get(1-i);
                            Collection<MetaBean> col = map.values();
                            List list = new ArrayList(col);
                            String json = JSON.serializeToString(list);
                            map.clear();

                            try(SendClient client = new SendClient(url)){
                                client.sendJSON(json);
                            }catch (Exception e){
                                e.printStackTrace();
                            }
                        }
                    }
                },
                0,
                1,
                TimeUnit.SECONDS);
    }

    public void put(String key, MetaBean bean){
        currentMap.put(key,bean);
    }
}
